/*
 * Copyright 2011-2014 the original author or authors.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.dianwoba.universe.redis.operation.impl;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import com.dianwoba.universe.cache.JedisCallback;
import com.dianwoba.universe.cache.JedisTemplate;
import com.dianwoba.universe.redis.operation.SetOperations;

import redis.clients.jedis.BinaryJedisCommands;

/**
 * Default implementation of {@link SetOperations}.
 * 
 */
public class DefaultSetOperations<K, V, X extends BinaryJedisCommands> 
    extends AbstractOperations<K, V, X> implements SetOperations<K, V> {

	public DefaultSetOperations(JedisTemplate<K, V, X> template) {
		super(template);
	}

	public Long add(K key, V... values) {
		final byte[] rawKey = rawKey(key);
		final byte[][] rawValues = rawValues(values);
		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.sadd(rawKey, rawValues);
			}
		});
	}

	public Boolean isMember(K key, Object o) {
		final byte[] rawKey = rawKey(key);
		final byte[] rawValue = rawValue(o);
		return execute(new JedisCallback<Boolean, X>() {

			public Boolean doInRedis(X jedis) {
				return jedis.sismember(rawKey, rawValue);
			}
		});
	}

	public Set<V> members(K key) {
		final byte[] rawKey = rawKey(key);
		Set<byte[]> rawValues = execute(new JedisCallback<Set<byte[]>, X>() {

			public Set<byte[]> doInRedis(X jedis) {
				return jedis.smembers(rawKey);
			}
		});

		return deserializeValues(rawValues);
	}

    @Override
    public Set<V> members(K key, Class<V> clazz) {
        final byte[] rawKey = rawKey(key);
        Set<byte[]> rawValues = execute(new JedisCallback<Set<byte[]>, X>() {

            public Set<byte[]> doInRedis(X jedis) {
                return jedis.smembers(rawKey);
            }
        });
        List<V> resultList = deserializeValues(new ArrayList<byte[]>(rawValues),clazz);
        return new HashSet<V>(resultList);
    }


    public V randomMember(K key) {

		return execute(new ValueDeserializingRedisCallback(key) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.srandmember(rawKey);
			}
		});
	}

	public Long remove(K key, Object... values) {
		final byte[] rawKey = rawKey(key);
		final byte[][] rawValues = rawValues(values);
		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.srem(rawKey, rawValues);
			}
		});
	}

	public V pop(K key) {
		return execute(new ValueDeserializingRedisCallback(key) {

			protected byte[] inRedis(byte[] rawKey, X jedis) {
				return jedis.spop(rawKey);
			}
		});
	}

	public Long size(K key) {
		final byte[] rawKey = rawKey(key);
		return execute(new JedisCallback<Long, X>() {

			public Long doInRedis(X jedis) {
				return jedis.scard(rawKey);
			}
		});
	}

}
